/**
 * @Author: hiram
 * @Date: 2021/12/10 下午 01:24
 */
package main

import (
	"gitee.com/gbat/nats-demo/common"
	"github.com/nats-io/nats.go"
	"log"
	"sync"
	"time"
)

func main() {
	var wg sync.WaitGroup
	wg.Add(1)
	for i := 0; i < 1; i++ {
		opts := []nats.Option{nats.Name("NATS Sample Queue Subscriber")}
		opts = common.SetupConnOptions(opts)
		// Connect to NATS
		nc, err := nats.Connect(nats.DefaultURL, opts...)
		if err != nil {
			log.Fatalln(err)
		}

		// Create JetStream Context
		js, err := nc.JetStream(nats.PublishAsyncMaxPending(256))
		if err != nil {
			log.Fatalln(err)
		}
		//Add stream
		//_, err= js.AddStream(&nats.StreamConfig{
		//    Name:     common.DefaultStreamName,
		//    Subjects: []string{common.DefaultSubject},
		//})
		//if err != nil {
		//    log.Fatalln(err)
		//}

		//Queue
		qsubDurable := nats.Durable("qdur-chan4") //消费者标识与subject保持一对一
		mch := make(chan *nats.Msg, 1)
		sub, err := js.ChanQueueSubscribe(common.DefaultSubject, common.DefaultQueue, mch, qsubDurable) //有序
		//sub, err := js.ChanSubscribe(common.DefaultSubject,  mch, nats.OrderedConsumer(),nats.DeliverNew()) //只获取连接后发送的新得消息  有序
		//sub, err := js.ChanSubscribe(common.DefaultSubject,  mch,nats.DeliverAll())

		if err != nil {
			log.Fatalf("Unexpected error: %v \n", err)
		}
		defer sub.Unsubscribe()

		go func(i int) {
			//var a *nats.MsgMetadata
			for {
				select {
				case msg := <-mch:
					//meta, err := msg.Metadata()
					//if err != nil {
					//    log.Fatalln(err)
					//}
					//a = meta
					err = msg.AckSync(nats.AckWait(10 * time.Second))
					if err != nil {
						log.Fatalln(err)
					}
					common.PrintMsg(msg, i)
				case <-time.After(20 * time.Second):
					log.Fatalln("Timeout waiting for message")
					wg.Done()
				}
			}
		}(i)

		//Simple Stream Publisher
		//sub,err:=js.ChanQueueSubscribe(common.DefaultSubject,common.DefaultQueue,func(msg *nats.Msg){//common.DefaultQueue
		//   common.PrintMsg(msg,0)
		//}, nats.DeliverLast())

		//sub, err:= js.SubscribeSync(common.DefaultSubject,nats.DeliverLast())
		//if err!=nil{
		//    log.Fatalln(err)
		//}
		//msg,err:=sub.NextMsg(10*time.Second)
		//if err!=nil{
		//    log.Fatalln(err)
		//}
		//common.PrintMsg(msg,i)
		//_, err= js.Publish(common.DefaultSubject, []byte(string(msg.Data)+strconv.Itoa(i)))
		//if err!=nil{
		//    log.Fatalln(err)
		//}
		//nats.DeliverNew()  不消费旧消息
		//nats.DeliverLast() 消费最近一条消息
		//nats.DeliverAll()  默认，消费所有消息，不分新旧
		//nats.OrderedConsumer() 顺序消费
		//if err != nil {
		//	log.Fatalln(err)
		//}
	}
	wg.Wait()
	//time.Sleep(time.Hour)
	//sb, _ := js.SubscribeSync(common.DefaultSubject, nats.OrderedConsumer(), nats.DeliverNew())
	//sb.NextMsg()
}
